package com.alison.tableapisql.chapter2_timeAttr;

import com.alison.tableapisql.chapter1_tableapiandsql.model.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class E1_TableTest5_TimeAndWindow {
    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String filePath = "D:\\workspace\\lab\\learnbigdata\\learnflink\\flink-datastream\\src\\main\\resources\\tableapi\\E1.txt";
        // 2. 读入文件数据，得到 DataStream
        DataStream<String> inputStream = env.readTextFile(filePath);

        // 3. 转换成POJO
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        // 4. 将流转换成表，定义时间特性
        Table dataTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp, pt.proctime");

        dataTable.printSchema();
        tableEnv.toAppendStream(dataTable, Row.class).print();

        env.execute();
    }
    /*
(
  `id` STRING,
  `ts` BIGINT,
  `temp` DOUBLE,
  `pt` TIMESTAMP_LTZ(3) *PROCTIME*
)

+I[sensor_1, 1547718199, 35.8, 2024-04-12T10:05:36.457Z]
+I[sensor_6, 1547718201, 15.4, 2024-04-12T10:05:36.459Z]
+I[sensor_7, 1547718202, 6.7, 2024-04-12T10:05:36.459Z]
+I[sensor_10, 1547718205, 38.1, 2024-04-12T10:05:36.460Z]
+I[sensor_1, 1547718207, 36.3, 2024-04-12T10:05:36.460Z]
+I[sensor_1, 1547718209, 32.8, 2024-04-12T10:05:36.460Z]
+I[sensor_1, 1547718212, 37.1, 2024-04-12T10:05:36.461Z]
     */
}